package impl

import (
	"context"
	"errors"
)

// 数据消费处理逻辑
func (i *impl) ConsumerEvent() {
	i.log.Debug().Msg("start consume event...")

	for {
		m, err := i.kr.FetchMessage(context.Background())
		if err != nil {
			i.log.Error().Msgf("consume event error, %s", err)
			break
		}
		i.log.Debug().Msgf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))

		err = i.kr.CommitMessages(context.Background(), m)
		if err != nil {
			i.log.Error().Msgf("commit error, %s", err)
		}

		// 存储事件 i.SaveEvent(), 模拟存储失败报错
		err = errors.New("save event error")
		if err != nil {
			i.colector.Inc()
		}
	}
}
